package com.examples.rabbitmqapi;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeoutException;

/**
 * @author benjamin
 * created at 2020-01-26
 */
public class Consumer {

    /**
     * 1、创建连接工厂
     * 2、创建连接
     * 3、创建信道
     * 4、声明交换器
     * 5、声明队列
     * 6、绑定队列： 用路由键绑定交换器和队列
     * 7、消费消息、确认消息
     */

    public static void main(String[] args) {

        ConnectionFactory factory = new ConnectionFactory();

        factory.setUsername("guest");
        factory.setPassword("guest");

        factory.setHost("localhost");
        factory.setVirtualHost("/");

        try {
            Connection connection = factory.newConnection();

            Channel channel = connection.createChannel();


            String exchange01 = "exchange01";

            // 交换器持久化
            channel.exchangeDeclare(exchange01, "direct", true);

            // queue声明
            String queue01 = channel.queueDeclare(
                    "queue01", true,
                    false, false, null).getQueue();


            String routingKey01 = "routingKey01";


            // 消费者制定绑定方法； 创建队列；
            // 1、 消费者创建队列之后； 生产者才能发布到队列
            //2、 destination队列是由消费者创建的
            channel.queueBind(queue01, exchange01, routingKey01);

            // 每次最多取1个
            channel.basicQos(1, true);


            DefaultConsumer callback = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                    try {
                        String msg = new String(body);

                        System.out.println("consumer: " + msg);
                    } finally {
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }

                }
            };



            // 间隔1s发起一次消费请求
            new Timer().schedule(new TimerTask() {
                @Override
                public void run() {

                    try {
                        String consume = channel.basicConsume(queue01, false, callback);
                        System.out.println("consume tag:  " + consume);

                    } catch (IOException e) {
                        e.printStackTrace();
                    }

                }
            }, 0L, 3000L);

        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {

        }
    }
}
